package com.study.flink.java.day08_tableAPI;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * Flink Table API示例  批处理
 */
public class BatchSqlWordCount {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);

        DataSource<WordCountEntity> input = env.fromElements(
                new WordCountEntity("spark", 1L),
                new WordCountEntity("spark", 1L),
                new WordCountEntity("hadoop", 1L),
                new WordCountEntity("hadoop", 1L),
                new WordCountEntity("hue", 1L),
                new WordCountEntity("hadoop", 1L),
                new WordCountEntity("hadoop", 1L),
                new WordCountEntity("flink", 1L),
                new WordCountEntity("flink", 1L),
                new WordCountEntity("flink", 1L));
        // 通过DataSet创建表
        Table table = tEnv.fromDataSet(input);
        Table filtered = table.groupBy("word") // 分组
                .select("word, counts.sum as counts") // sum
                .filter("counts >= 2") // 过滤
                .orderBy("counts.desc"); //排序
        DataSet<WordCountEntity> result = tEnv.toDataSet(filtered, WordCountEntity.class);
        result.print();
        //env.execute("BatchSqlWordCount");
    }


}
